package com.test.study.ToolsUtils.strom.trident;

import java.util.HashMap;
import java.util.Map;

import org.apache.storm.trident.operation.BaseAggregator;
import org.apache.storm.trident.operation.TridentCollector;
import org.apache.storm.trident.tuple.TridentTuple;
import org.apache.storm.tuple.Values;

public class MyAggregator extends BaseAggregator<Map<String, Integer>> {
	/**
		 * 
		 */
	private static final long serialVersionUID = -2856606128918200307L;

	/**
	 * 初始化方法
	 */
	@Override
	public Map<String, Integer> init(Object batchId, TridentCollector collector) {
		return new HashMap<String, Integer>();
	}

	/**
	 * 具体的业务逻辑 参数1:初始化的HashMap
	 */

	@Override
	public void aggregate(Map<String, Integer> val, TridentTuple tuple, TridentCollector collector) {
		String word = tuple.getString(0);
		Integer value = val.get(word);
		if (value == null) {
			value = 0;
		}
		value++;
		val.put(word, value);

	}

	/**
	 * 将这个数据发出去 这是个局部汇总
	 */
	@Override
	public void complete(Map<String, Integer> val, TridentCollector collector) {
		collector.emit(new Values(val));

	}

}
